package com.tuozixuan.flink.demo;

import java.util.Calendar;
import java.util.Random;

import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

public class SensorSource extends RichParallelSourceFunction<SensorReading> {

	private static final long serialVersionUID = -700177256698094453L;
	
	private boolean running = true;
	 
	@Override
	public void run(SourceContext<SensorReading> ctx) throws Exception {

        Random rand = new Random();

        String[] sensorIds = new String[10];
        double[] curFTemp = new double[10];
        for (int i = 0; i < 10; i++) {
            sensorIds[i] = "sensor_" + i;
            curFTemp[i] = 65 + (rand.nextGaussian() * 20);
        }

        while (running) {
            long curTime = Calendar.getInstance().getTimeInMillis();
            for (int i = 0; i < 10; i++) {
                curFTemp[i] += rand.nextGaussian() * 0.5;
                ctx.collect(new SensorReading(sensorIds[i], curTime, curFTemp[i]));
            }

            Thread.sleep(100);
        }
	}

	@Override
	public void cancel() {
		running = false;
	}

}
